package cn.doitedu.dw_etl

import cn.doitedu.dw_beans.ApplogBean
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-12-13
 * @desc 给ods的app行为日志补充用户唯一标识
 * 总体思路：
 *    从一条日志中取出  device_id 和 登录账号
 *       如果有 登录账号，则从 用户注册信息表中查询它的guid
 *       如果无 登录账号，则用device_id 去 设备账号绑定权重表 查对应的绑定账号，然后再用账号查 guid
 *       如果在设备账号绑定权重表中依然查不到账号，则用device_id去 空设备guid映射表 中查guid
 */
object ApplogOds2DwdGuid {
  def main(args: Array[String]): Unit = {
    if(args.size<1){
      println(
        """
          |
          |params not enough!
          |usage:
          |  args(0): 要计算的分区日期
          |
          |""".stripMargin)
      sys.exit(1)
    }

    val dt = args(0)


    val spark: SparkSession = SparkSession.builder()
      .appName("用户唯一标识")
      //.master("local")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._


    // 读上一步（地理位置集成）的结果
    val ds: Dataset[ApplogBean] = spark.read.parquet(s"hdfs://doit01:8020/tmp/tasktmp/areaAdded/${dt}/").as[ApplogBean]
    ds.createTempView("log_dtl")


    // 读取hive中的  设备账号绑定表， 并筛选出每个设备绑定关系最强的那一个账号
    // d01,c01,1000,t2
    // d01,c02,800,t3
    val device_account_bind = spark.sql(
      s"""
        |select
        |  device_id,
        |  account
        |from (
        |select
        |  device_id,
        |  account,
        |  row_number() over(partition by device_id order by weight desc,last_login_time desc) as rn
        |from dim.dev_acc_bind_w
        |where dt='${dt}'
        |) o
        |where rn=1
        |""".stripMargin)
    device_account_bind.createTempView("device_account_bind")


    // 先对日志数据缺失 账号的 数据 ，补上 动态绑定的最强账号
    val accountMerge = spark.sql(
      """
        |
        |select
        |  -- （日志中没有账号，而绑定表中关联上了账号），才用关联上的账号
        |  if(a.account is null and b.account is not null,b.account,a.account) as account,
        |  a.appid,
        |  a.appversion,
        |  a.carrier,
        |  a.deviceid,
        |  a.devicetype,
        |  a.eventid,
        |  a.ip,
        |  a.latitude,
        |  a.longitude,
        |  a.nettype,
        |  a.osname,
        |  a.osversion,
        |  a.properties,
        |  a.releasechannel,
        |  a.resolution,
        |  a.sessionid,
        |  a.timestamp,
        |  a.splitedSessionId,
        |  a.province,
        |  a.city,
        |  a.district
        |
        |from log_dtl a left  join  device_account_bind b on a.deviceid = b.device_id
        |
        |""".stripMargin)
    accountMerge.createTempView("account_merged")

    // 将填补账号后，确实有账号的数据  关联 用户注册信息表  ，得到guid
    spark.sql(
      s"""
        |
        |insert into table dwd.mall_app_event_dtl partition(dt='${dt}')
        |
        |select
        |  a.account,
        |  a.appid,
        |  a.appversion,
        |  a.carrier,
        |  a.deviceid,
        |  a.devicetype,
        |  a.eventid,
        |  a.ip,
        |  a.latitude,
        |  a.longitude,
        |  a.nettype,
        |  a.osname,
        |  a.osversion,
        |  a.properties,
        |  a.releasechannel,
        |  a.resolution,
        |  a.sessionid,
        |  a.timestamp,
        |  a.splitedSessionId,
        |  a.province,
        |  a.city,
        |  a.district,
        |  b.user_id  as guid
        |from account_merged a join dim.user_info b  on a.account=b.account
        |
        |union all
        |
        |select
        |  o1.account,
        |  o1.appid,
        |  o1.appversion,
        |  o1.carrier,
        |  o1.deviceid,
        |  o1.devicetype,
        |  o1.eventid,
        |  o1.ip,
        |  o1.latitude,
        |  o1.longitude,
        |  o1.nettype,
        |  o1.osname,
        |  o1.osversion,
        |  o1.properties,
        |  o1.releasechannel,
        |  o1.resolution,
        |  o1.sessionid,
        |  o1.timestamp,
        |  o1.splitedSessionId,
        |  o1.province,
        |  o1.city,
        |  o1.district,
        |  o2.guid  as guid
        |from
        |(
        |   select
        |     *
        |   from account_merged
        |   where account is null
        |) o1
        |left join
        |dim.dev_guid_map o2
        |on o1.deviceid = o2.device_id
        |
        |
        |""".stripMargin)


    spark.close()

  }
}
